package com.bw.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.app.func.DimAsyncFunction;
import com.bw.gmall.realtime.bean.TradeTrademarkCategoryUserRefundBean;
import com.bw.gmall.realtime.utils.AsyncDimFunction;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class Dwd_Shop_NewVisitUser {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> dataStreamSource = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log", "dwd_traffic_page_log_dwd_split"));
        SingleOutputStreamOperator<JSONObject> map = dataStreamSource.map(line -> JSONObject.parseObject(line));
        SingleOutputStreamOperator<JSONObject> filter = map.filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                String string = value.getJSONObject("page").getString("item_type");
                if ("sku_id".equals(string) || "sku_ids".equals(string) || "sku_id".equals(string)) {
                    return true;
                }
                return false;
            }
        });
//        filter.print();  1，2
        SingleOutputStreamOperator<JSONObject> flatMap = filter.flatMap(new FlatMapFunction<JSONObject, JSONObject>() {
            @Override
            public void flatMap(JSONObject value, Collector<JSONObject> out) throws Exception {
                // 获取page中的item_type
                String itemType = value.getJSONObject("page").getString("item_type");
                // 仅处理item_type为"sku_ids"的情况
                if ("sku_ids".equals(itemType)) {
                    // 拆分item为数组（如"1,2"拆分为["1","2"]）
                    String[] splitItems = value.getJSONObject("page").getString("item").split(",");

                    // 遍历拆分后的每个元素，生成新的JSONObject并输出
                    for (String item : splitItems) {
                        // 复制原JSONObject（避免修改原对象导致的引用问题）
                        JSONObject newJson = JSONObject.parseObject(value.toString());
                        // 设置当前item到page中
                        newJson.getJSONObject("page").put("item", item);
                        // 将新对象收集到输出流中
                        out.collect(newJson);
                    }
                } else {
                    // 非目标类型，直接输出原对象
                    out.collect(value);
                }
            }
        });
//        flatMap.print();
        SingleOutputStreamOperator<JSONObject> ts = flatMap.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong("ts");
            }
        }));
//     ts.print();

        // 使用更高的并行度和更长的超时时间
        SingleOutputStreamOperator<JSONObject> jsonObjectSingleOutputStreamOperator = AsyncDataStream.unorderedWait(ts, new DimAsyncFunction<JSONObject>("DIM_SKU_INFO_ZYJ") {
            @Override
            public String getKey(JSONObject input) {
                return input.getJSONObject("page").getString("item");
            }

            @Override
            public void join(JSONObject input, JSONObject dimInfo) throws Exception {
                input.getJSONObject("page").put("shop", dimInfo.getString("SHOP_ID"));
            }
        }, 100, TimeUnit.SECONDS);
//        jsonObjectSingleOutputStreamOperator.print("aaaaaaaaaaaaa----------->");
        SingleOutputStreamOperator<JSONObject> filter1 = jsonObjectSingleOutputStreamOperator.filter(jsonObject -> jsonObject.getJSONObject("page").getString("shop") != null);
        KeyedStream<JSONObject, Tuple2<String, String>> keyBy = filter1.keyBy(new KeySelector<JSONObject, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(JSONObject value) throws Exception {
                String shop = value.getJSONObject("page").getString("shop");
                String mid = value.getJSONObject("common").getString("mid");
                return Tuple2.of(mid, shop);
            }
        });

        SingleOutputStreamOperator<JSONObject> process = keyBy.process(new ProcessFunction<JSONObject, JSONObject>() {
            private ValueState<String> shopState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor =
                        new ValueStateDescriptor<>("last_visit_dt", String.class);
                shopState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                //店铺新访客状态新增
                String shop = value.getJSONObject("page").getString("shop");
                String value1 = shopState.value();
                if (value1 == null) {
                    shopState.update(shop);
                    value.getJSONObject("page").put("shop_new_visit", "1");
                    out.collect(value);
                } else {
                    value.getJSONObject("page").put("shop_new_visit", "0");
                    out.collect(value);
                }

            }
        });
        process.print("shop_new_visit>>>>>>>>>>>>>>");
        process.map(t->t.toJSONString()).addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_shop_new_visit"));
        env.execute();
    }
}
